package org.jeecg.modules.kafka;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import java.util.Date;
import java.util.concurrent.atomic.AtomicBoolean;


//@Component
@Slf4j
public class KafkaDeviceSendConsumer {

    private final AtomicBoolean listening = new AtomicBoolean(false);

    /**
     * 消费者配置，kafka.topic为监听的topic，kafka.group为消费分组，可在yml中修改
     */
//    @KafkaListener(topics = "ig2apiDeviceInfo_13Sk5deg7JH55ZmY_CWxK153mic4aJhijVXoxg1UsFF1NeaSc")
//    @Async
    public void kafkaListener(ConsumerRecord<String, String> consumerRecord) {
        String value = consumerRecord.value();
        log.info("收到kafka消息ig2apiDeviceInfo：{}", value);
    }

//    @KafkaListener(topics = "ig2apiDataInfo_13Sk5deg7JH55ZmY_CWxK153mic4aJhijVXoxg1UsFF1NeaSc")
//    @Async
    public void kafkaListenerDeviceData(ConsumerRecord<String, String> consumerRecord) {
        if (!listening.get()) {
            // 如果没有开始监听，则直接返回
            return;
        }
        String value = consumerRecord.value();
        JSONObject jsonObject = JSONObject.parseObject(value);
        Date gpsTime = new Date(jsonObject.getLongValue("gpsTime"));
        String deviceCode = jsonObject.getString("deviceCode");
        JSONObject data = jsonObject.getJSONObject("data");
        String lng = data.getString("lng");
        String lat = data.getString("lat");
        log.info("经度lng：{}", lng);
        log.info("纬度lat：{}", lat);
        log.info("gps时间gpsTime：{}", gpsTime);
        log.info("设备编号deviceCode：{}", deviceCode);
    }

    public void startListening() {
        listening.set(true);
    }

    public void stopListening() {
        listening.set(false);
    }

}
